package org.acmclass.fatworm.test.sql;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Semaphore;

import org.acmclass.fatworm.test.Timer;
import org.acmclass.fatworm.test.plaintext.Log;
import org.acmclass.fatworm.test.result.ResultConsumer;

public class SQLExecutor implements Runnable {

	private int threadID;
	private String threadName;
	private String url, user, password;
	private List<SQLWithOptions> sqlList;
	private ResultConsumer consumer;
	private Log logger;
	private Timer timer = new Timer();
	private Semaphore semaphore = new Semaphore(0);

	public SQLExecutor(int threadNum, String url, String user, String password) {
		this.threadID = threadNum;
		this.threadName = "[Thread" + threadNum + "] ";
		this.url = url;
		this.user = user;
		this.password = password;
		sqlList = new LinkedList<SQLWithOptions>();
	}

	public void addSQL(SQLWithOptions sql) {
		sqlList.add(sql);
	}

	public void setResultConsumer(ResultConsumer consumer) {
		this.consumer = consumer;
	}

	public void setLogger(Log logger) {
		this.logger = logger;
	}

	public void waitFor() {
		try {
			semaphore.acquire();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	public long getTime() {
		return timer.getTime();
	}

	private void log(String msg) {
		logger.log(threadName, msg);
	}

	@Override
	public void run() {
		Connection connection = null;
		try {
			log("Connecting " + url + "?user=" + user + "&password=" + password);
			connection = DriverManager.getConnection(url, user, password);
			log("Connection established.");
			timer.start();

			for (SQLWithOptions sql : sqlList) {
				try {
					Statement statement = connection.createStatement();
					timer.pause();
					log("Executing: " + sql.getSql());
					timer.resume();
					boolean isQuery = statement.execute(sql.getSql());
					if (isQuery) {
						log("Query executed, begin to fetch and check result");
						ResultSet resultSet = statement.getResultSet();
						timer.pause();
						consumer.setSQL(threadID, sql);
						consumer.setResultSetMetaData(threadID, resultSet
								.getMetaData());
						timer.resume();
						int columnCount = resultSet.getMetaData()
								.getColumnCount();
						Map<Integer, Object> row = new HashMap<Integer, Object>();
						while (resultSet.next()) {
							row.clear();
							for (int i = 1; i <= columnCount; ++i) {
								row.put(i, resultSet.getObject(i));
							}
							timer.pause();
							consumer.process(threadID, row);
							timer.resume();
						}
						resultSet.close();
						timer.pause();
						consumer.processQueryFinal(threadID);
						timer.resume();
					}
					statement.close();
				} catch (SQLException e) {
					e.printStackTrace();
				}
			}

		} catch (SQLException e) {
			e.printStackTrace();
		} finally {
			timer.pause();
			if (connection != null) {
				try {
					connection.close();
				} catch (SQLException e) {
					e.printStackTrace();
				}
			}
			semaphore.release();
		}

	}

}
